#include <iostream>

#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_awaitable.hpp>

#include <boost/asio/ip/tcp.hpp>

#include <async_mqtt5.hpp>

#include <thread>

bool exitFlag=false;

using namespace std::chrono_literals;
#ifdef BOOST_ASIO_HAS_CO_AWAIT

namespace asio = boost::asio;

asio::awaitable<void> client_publisher(async_mqtt5::mqtt_client<asio::ip::tcp::socket>& client) {

	async_mqtt5::subscribe_topic sub_topic = async_mqtt5::subscribe_topic {
		"/supermatrix/test", //主题名称
		async_mqtt5::subscribe_options {
			async_mqtt5::qos_e::exactly_once, // All messages will arrive at QoS 2.
			async_mqtt5::no_local_e::no, // Forward message from Clients with same ID.
			async_mqtt5::retain_as_published_e::retain, // Keep the original RETAIN flag.
			async_mqtt5::retain_handling_e::send
		}
	};

	auto [sub_codes, sub_props] = co_await client.async_subscribe(
		sub_topic, async_mqtt5::subscribe_props {}, asio::use_awaitable
	);

  auto [rc, props] = co_await client.async_publish<async_mqtt5::qos_e::at_least_once>(
		"/supermatrix/test", "my application message",
		async_mqtt5::retain_e::yes, async_mqtt5::publish_props {}, asio::use_awaitable
	);
	if (rc)
		std::cout << "MQTT protocol error occurred: " << rc.message() << std::endl;

  std::cout<<"Enter Recieve await"<<std::endl;
  while(true)
  {
    try
    {
	    auto [topic, payload, publish_props] = co_await client.async_receive(asio::use_awaitable);
      std::cout<<"topic= "<<topic<<" payload= "<<payload<<std::endl;
    }
    catch(const async_mqtt5::error_code& e)
    {
      std::cout<<"Receive Exit Signal "<<e.what()<<std::endl;
      break;
    }
  }

	auto [unsub_codes, unsub_props] = co_await client.async_unsubscribe(
		"/supermatrix/test", async_mqtt5::unsubscribe_props {},
		asio::use_awaitable
	);

	co_await client.async_disconnect(
		async_mqtt5::disconnect_rc_e::disconnect_with_will_message,
    async_mqtt5::disconnect_props {}, 
    asio::use_awaitable
	);

	co_return;
}

asio::awaitable<void> timer(asio::io_context& ioc,async_mqtt5::mqtt_client<asio::ip::tcp::socket>& client)
{
  asio::steady_timer timer(ioc);
  timer.expires_after(1s);
  while(!exitFlag)
    co_await timer.async_wait(asio::use_awaitable);
  std::cout<<"Exit Flag set"<<std::endl;
  client.cancel();
  co_return;
}


int main() {
	// Initialise execution context.
	asio::io_context ioc;

	async_mqtt5::mqtt_client<asio::ip::tcp::socket> client(ioc);

  client
    .credentials("mqtt_test_cpp","supermatrix_test","supermatrix_test")
    .brokers("9f1a27fc-db77-e97a-35fa-8d0d602616a0.openfrp.cc", 59640)
		.async_run(asio::detached); // Start the client.

	// Spawn the coroutine.
	co_spawn(ioc.get_executor(), client_publisher(client), asio::detached);
  co_spawn(ioc.get_executor(),timer(ioc,client),asio::detached);

  std::thread exitThread([&](){
    std::this_thread::sleep_for(10s);
    exitFlag=true;
  });
	// Start the execution.
	ioc.run();
  exitThread.join();
}

#endif
